package org.jgroups.tests;

import org.jgroups.ReceiverAdapter;
import org.jgroups.JChannel;
import org.jgroups.Message;
import org.jgroups.stack.Protocol;
import org.jgroups.util.Streamable;
import org.jgroups.util.Util;
import org.jgroups.jmx.JmxConfigurator;
import org.jgroups.protocols.FragHeader3;
import org.jgroups.protocols.FRAG2;
import org.jgroups.protocols.pbcast.NAKACK;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.Vector;

import javax.management.MBeanServer;
import javax.management.MBeanServerFactory;

/**
 * Created by IntelliJ IDEA.
 * User: thsieh
 * Date: May 11, 2009
 * Time: 10:15:00 AM
 * To change this template use File | Settings | File Templates.
 */
public class TestJSender extends ReceiverAdapter {
    private static Log logger = LogFactory.getLog(TestJSender.class);
   private  static FileOutputStream sentLog;
    private static FileOutputStream recvLog;
    static {
    	try {
    	sentLog = new FileOutputStream(new File("sent.log"),true);
    	recvLog = new FileOutputStream(new File("recv.log"),true);
    	} catch(FileNotFoundException fnfe) {
    		logger.error("msgs.log cannot be opened");
    	}
    }

    public TestJSender() {

    }
    public void receive(Message msg) {
        if ( msg == null ) {
            logger.warn("Mesasge is null ");
        }
        else {
            FragHeader3 header = (FragHeader3) msg.getHeader("FRAG3") ;
            logger.info("Receive message source=" +msg.getSrc().toString()+" length="+ msg.getLength()
                    +" destination="+ msg.getDest());
        }
    }

    public static void help() {
          System.out.println("java stress.org.jgroups.test.TestSender -udp ./udp.xml -cluster test-1 -count 100 size -size 10");
    }
    public static class MessagePayload implements Streamable {
    	public byte[] data;
    	public static AtomicLong idCtr = new AtomicLong(0);
    	public long id;
    	public MessagePayload(byte[] data) {
    		id = idCtr.getAndIncrement();
    		this.data = data;
    	}
    	
    	public MessagePayload() {}
    	
		@Override
		public void readFrom(DataInputStream in) throws IOException,
				IllegalAccessException, InstantiationException {
			id = in.readLong();
			int dataSize = in.readInt();
			data = new byte[dataSize];
			in.read(data);
		}
		@Override
		public void writeTo(DataOutputStream out) throws IOException {
			out.writeLong(id);
			out.writeInt(data.length);
			out.write(data);
		}
    	}
    	
    public static NAKACK getNAKACK(JChannel channel) {
        Vector<Protocol> vectors = channel.getProtocolStack().getProtocols();
        for (Protocol p : vectors ) {
            if ( p.getName().equals("NAKACK")) return  (NAKACK) p;
        }
        return null;
    }


    public static void main(String[] args) throws Exception {

        String[] defaults = { "./udp.xml","test-1" , "100" , "20", "2000", "3" };

        for ( int i=0; i < args.length ; ) {
            if ( args[i].equals("-udp")) defaults[0] = args[++i] ;
            else if ( args[i].equals("-cluster")) defaults[1] = args[++i] ;
            else if ( args[i].equals("-count")) defaults[2] = args[++i] ;
            else if ( args[i].equals("-size")) defaults[3] = args[++i] ;
            else if ( args[i].equals("-times")) defaults[4] = args[++i] ;
            else if ( args[i].equals("-sleep")) defaults[5] = args[++i] ;
            else if ( args[i].equals("-help")) help() ;
            else {
               help();
            }
            i ++;
        }
        TestJSender ts = new TestJSender();
        System.setProperty("java.net.preferIPv4Stack", "true");
        logger.info("java.net.preferIPv4Stack ="+ System.getProperty("java.net.preferIPv4Stack"));

        JChannel channel = new JChannel(new File(defaults[0]));
        channel.setReceiver(ts);
        channel.connect(defaults[1]);

 
        MBeanServer server=MBeanServerFactory.createMBeanServer();
        
        JmxConfigurator.registerChannel(channel, server, "JGroups"+  channel.getClusterName() , null, true);
        
        int count = Integer.valueOf( defaults[2]);
        int size  = Integer.valueOf( defaults[3]);
        int times = Integer.valueOf( defaults[4]);
        int sleep = Integer.valueOf( defaults[5]);;
        FRAG2 frag2 = (FRAG2 ) channel.getProtocolStack().getTopProtocol();
        NAKACK nakack = getNAKACK( channel);
        for (int j= 0; j < times ; j ++ ) {
            logger.info("Begin iteration " + j );
            long begin = System.nanoTime();
            for (int i = 0;  i < count ; i ++) {
                int s = ( i % size ) +1 ;
                MessagePayload mpl = new MessagePayload(new byte[59800* s]);
                Message msg = new Message(null, null, Util.objectToByteBuffer(mpl) );
                channel.send( msg);
                sentLog.write((Long.toString(mpl.id) + "\n").getBytes());
                sentLog.flush();
            }
            long end = System.nanoTime();
            logger.info("Iteration "+ j +"  duration micro seconds "+ TimeUnit.MICROSECONDS.convert((end - begin), TimeUnit.NANOSECONDS ));
            logger.info("End iteration " + j +" and sleep for "+ sleep + " seconds" );
            Thread.sleep( 1000L * sleep );
            logger.info("FRAG2 "+ frag2.toString());
            logger.info("NAKACK " + nakack.toString());
        }

    }



    
}
